using System.Collections.Concurrent;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using StackExchange.Redis;

namespace SumerCoreDevOps.Infrastructure.Distributed;

/// <summary>
/// 基于 Redis 的分布式协调器实现
/// </summary>
public class RedisCoordinator : IDistributedCoordinator, IAsyncDisposable
{
    private readonly IConnectionMultiplexer _redis;
    private readonly IDatabase _database;
    private readonly ISubscriber _subscriber;
    private readonly ILogger<RedisCoordinator> _logger;
    private readonly string _keyPrefix;
    private readonly ConcurrentDictionary<string, Action<RedisChannel, RedisValue>> _subscriptions = new();

    public RedisCoordinator(
        IConnectionMultiplexer redis,
        ILogger<RedisCoordinator> logger,
        string keyPrefix = "sumer:")
    {
        _redis = redis ?? throw new ArgumentNullException(nameof(redis));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _keyPrefix = keyPrefix;
        _database = _redis.GetDatabase();
        _subscriber = _redis.GetSubscriber();
    }

    private string GetKey(string key) => $"{_keyPrefix}{key}";

    #region Lock Operations

    public async Task<IDistributedLock?> AcquireLockAsync(string key, TimeSpan timeout, CancellationToken cancellationToken = default)
    {
        var lockKey = GetKey($"lock:{key}");
        var token = Guid.NewGuid().ToString("N");
        var endTime = DateTime.UtcNow.Add(timeout);

        while (DateTime.UtcNow < endTime && !cancellationToken.IsCancellationRequested)
        {
            var acquired = await _database.StringSetAsync(
                lockKey,
                token,
                timeout,
                When.NotExists);

            if (acquired)
            {
                _logger.LogDebug("Acquired distributed lock: {Key} with token: {Token}", key, token);
                return new RedisDistributedLock(this, lockKey, token, _logger);
            }

            await Task.Delay(100, cancellationToken);
        }

        _logger.LogWarning("Failed to acquire distributed lock: {Key} within timeout: {Timeout}", key, timeout);
        return null;
    }

    public async Task<IDistributedLock?> TryAcquireLockAsync(string key, TimeSpan timeout, CancellationToken cancellationToken = default)
    {
        var lockKey = GetKey($"lock:{key}");
        var token = Guid.NewGuid().ToString("N");

        var acquired = await _database.StringSetAsync(
            lockKey,
            token,
            timeout,
            When.NotExists);

        if (acquired)
        {
            _logger.LogDebug("Acquired distributed lock (try): {Key} with token: {Token}", key, token);
            return new RedisDistributedLock(this, lockKey, token, _logger);
        }

        return null;
    }

    internal async Task<bool> ReleaseLockAsync(string lockKey, string token)
    {
        const string script = @"
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end";

        var result = await _database.ScriptEvaluateAsync(
            script,
            new RedisKey[] { lockKey },
            new RedisValue[] { token });

        var released = (int)result == 1;
        if (released)
        {
            _logger.LogDebug("Released distributed lock: {Key}", lockKey);
        }
        else
        {
            _logger.LogWarning("Failed to release distributed lock: {Key} (token mismatch or expired)", lockKey);
        }

        return released;
    }

    internal async Task<bool> ExtendLockAsync(string lockKey, string token, TimeSpan expiry)
    {
        const string script = @"
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('pexpire', KEYS[1], ARGV[2])
            else
                return 0
            end";

        var result = await _database.ScriptEvaluateAsync(
            script,
            new RedisKey[] { lockKey },
            new RedisValue[] { token, (long)expiry.TotalMilliseconds });

        return (int)result == 1;
    }

    #endregion

    #region Key-Value Operations

    public async Task SetAsync<T>(string key, T value, TimeSpan? expiry = null, CancellationToken cancellationToken = default)
    {
        var redisKey = GetKey(key);
        var serialized = JsonSerializer.Serialize(value);
        await _database.StringSetAsync(redisKey, serialized, expiry);
        _logger.LogDebug("Set key: {Key}, expiry: {Expiry}", key, expiry);
    }

    public async Task<T?> GetAsync<T>(string key, CancellationToken cancellationToken = default)
    {
        var redisKey = GetKey(key);
        var value = await _database.StringGetAsync(redisKey);

        if (value.IsNullOrEmpty)
        {
            return default;
        }

        try
        {
            return JsonSerializer.Deserialize<T>(value.ToString());
        }
        catch (JsonException ex)
        {
            _logger.LogError(ex, "Failed to deserialize value for key: {Key}", key);
            return default;
        }
    }

    public async Task<bool> DeleteAsync(string key, CancellationToken cancellationToken = default)
    {
        var redisKey = GetKey(key);
        var deleted = await _database.KeyDeleteAsync(redisKey);
        _logger.LogDebug("Delete key: {Key}, result: {Result}", key, deleted);
        return deleted;
    }

    public async Task<long> DeletePatternAsync(string pattern, CancellationToken cancellationToken = default)
    {
        var redisPattern = GetKey(pattern);
        var server = _redis.GetServer(_redis.GetEndPoints().First());
        var keys = server.Keys(pattern: redisPattern).ToArray();

        if (keys.Length == 0)
        {
            return 0;
        }

        var deleted = await _database.KeyDeleteAsync(keys);
        _logger.LogDebug("Deleted {Count} keys matching pattern: {Pattern}", deleted, pattern);
        return deleted;
    }

    public async Task<bool> ExistsAsync(string key, CancellationToken cancellationToken = default)
    {
        var redisKey = GetKey(key);
        return await _database.KeyExistsAsync(redisKey);
    }

    public async Task<bool> ExpireAsync(string key, TimeSpan expiry, CancellationToken cancellationToken = default)
    {
        var redisKey = GetKey(key);
        return await _database.KeyExpireAsync(redisKey, expiry);
    }

    #endregion

    #region Pub/Sub Operations

    public async Task<long> PublishAsync<T>(string channel, T message, CancellationToken cancellationToken = default)
    {
        var redisChannel = GetKey(channel);
        var serialized = JsonSerializer.Serialize(message);
        var subscribers = await _subscriber.PublishAsync(redisChannel, serialized);
        _logger.LogDebug("Published message to channel: {Channel}, subscribers: {Subscribers}", channel, subscribers);
        return subscribers;
    }

    public async Task SubscribeAsync<T>(string channel, Func<T, Task> handler, CancellationToken cancellationToken = default)
    {
        var redisChannel = GetKey(channel);

        Action<RedisChannel, RedisValue> redisHandler = async (ch, value) =>
        {
            try
            {
                var message = JsonSerializer.Deserialize<T>(value.ToString());
                if (message != null)
                {
                    await handler(message);
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing message from channel: {Channel}", channel);
            }
        };

        _subscriptions[channel] = redisHandler;
        await _subscriber.SubscribeAsync(redisChannel, redisHandler);
        _logger.LogInformation("Subscribed to channel: {Channel}", channel);
    }

    public async Task UnsubscribeAsync(string channel, CancellationToken cancellationToken = default)
    {
        var redisChannel = GetKey(channel);

        if (_subscriptions.TryRemove(channel, out var handler))
        {
            await _subscriber.UnsubscribeAsync(redisChannel, handler);
            _logger.LogInformation("Unsubscribed from channel: {Channel}", channel);
        }
    }

    #endregion

    #region Hash Operations

    public async Task HashSetAsync<T>(string key, string field, T value, CancellationToken cancellationToken = default)
    {
        var redisKey = GetKey(key);
        var serialized = JsonSerializer.Serialize(value);
        await _database.HashSetAsync(redisKey, field, serialized);
        _logger.LogDebug("Hash set: {Key}.{Field}", key, field);
    }

    public async Task<T?> HashGetAsync<T>(string key, string field, CancellationToken cancellationToken = default)
    {
        var redisKey = GetKey(key);
        var value = await _database.HashGetAsync(redisKey, field);

        if (value.IsNullOrEmpty)
        {
            return default;
        }

        try
        {
            return JsonSerializer.Deserialize<T>(value.ToString());
        }
        catch (JsonException ex)
        {
            _logger.LogError(ex, "Failed to deserialize hash value for key: {Key}.{Field}", key, field);
            return default;
        }
    }

    public async Task<Dictionary<string, T>> HashGetAllAsync<T>(string key, CancellationToken cancellationToken = default)
    {
        var redisKey = GetKey(key);
        var entries = await _database.HashGetAllAsync(redisKey);
        var result = new Dictionary<string, T>();

        foreach (var entry in entries)
        {
            try
            {
                var value = JsonSerializer.Deserialize<T>(entry.Value.ToString());
                if (value != null)
                {
                    result[entry.Name.ToString()] = value;
                }
            }
            catch (JsonException ex)
            {
                _logger.LogError(ex, "Failed to deserialize hash value for key: {Key}.{Field}", key, entry.Name);
            }
        }

        return result;
    }

    public async Task<bool> HashDeleteAsync(string key, string field, CancellationToken cancellationToken = default)
    {
        var redisKey = GetKey(key);
        var deleted = await _database.HashDeleteAsync(redisKey, field);
        _logger.LogDebug("Hash delete: {Key}.{Field}, result: {Result}", key, field, deleted);
        return deleted;
    }

    #endregion

    #region Master Node Management

    public async Task RegisterMasterAsync(string masterId, string host, int tcpPort, int apiPort, CancellationToken cancellationToken = default)
    {
        var masterKey = GetKey($"master:{masterId}");
        var masterInfo = new MasterInfo
        {
            MasterId = masterId,
            Host = host,
            TcpPort = tcpPort,
            ApiPort = apiPort,
            LastHeartbeat = DateTime.UtcNow,
            IsOnline = true
        };

        await SetAsync(masterKey, masterInfo, TimeSpan.FromSeconds(90), cancellationToken);
        _logger.LogInformation("Registered master node: {MasterId} at {Host}:{TcpPort}", masterId, host, tcpPort);
    }

    public async Task UpdateMasterHeartbeatAsync(string masterId, CancellationToken cancellationToken = default)
    {
        var masterKey = GetKey($"master:{masterId}");
        var masterInfo = await GetAsync<MasterInfo>(masterKey, cancellationToken);

        if (masterInfo != null)
        {
            masterInfo.LastHeartbeat = DateTime.UtcNow;
            masterInfo.IsOnline = true;
            await SetAsync(masterKey, masterInfo, TimeSpan.FromSeconds(90), cancellationToken);
            _logger.LogDebug("Updated master heartbeat: {MasterId}", masterId);
        }
    }

    public async Task<List<MasterInfo>> GetOnlineMastersAsync(CancellationToken cancellationToken = default)
    {
        var pattern = GetKey("master:*");
        var server = _redis.GetServer(_redis.GetEndPoints().First());
        var keys = server.Keys(pattern: pattern);

        var masters = new List<MasterInfo>();
        foreach (var key in keys)
        {
            var value = await _database.StringGetAsync(key);
            if (!value.IsNullOrEmpty)
            {
                try
                {
                    var masterInfo = JsonSerializer.Deserialize<MasterInfo>(value.ToString());
                    if (masterInfo != null && masterInfo.IsOnline)
                    {
                        masters.Add(masterInfo);
                    }
                }
                catch (JsonException ex)
                {
                    _logger.LogError(ex, "Failed to deserialize master info for key: {Key}", key);
                }
            }
        }

        return masters;
    }

    public async Task UnregisterMasterAsync(string masterId, CancellationToken cancellationToken = default)
    {
        var masterKey = GetKey($"master:{masterId}");
        await DeleteAsync(masterKey, cancellationToken);
        _logger.LogInformation("Unregistered master node: {MasterId}", masterId);
    }

    #endregion

    public async ValueTask DisposeAsync()
    {
        foreach (var channel in _subscriptions.Keys.ToArray())
        {
            await UnsubscribeAsync(channel);
        }
    }
}

/// <summary>
/// Redis 分布式锁实现
/// </summary>
internal class RedisDistributedLock : IDistributedLock
{
    private readonly RedisCoordinator _coordinator;
    private readonly ILogger _logger;
    private bool _disposed;

    public string Key { get; }
    public string Token { get; }
    public bool IsAcquired { get; private set; }

    public RedisDistributedLock(RedisCoordinator coordinator, string key, string token, ILogger logger)
    {
        _coordinator = coordinator;
        Key = key;
        Token = token;
        IsAcquired = true;
        _logger = logger;
    }

    public async Task<bool> ExtendAsync(TimeSpan expiry, CancellationToken cancellationToken = default)
    {
        if (_disposed || !IsAcquired)
        {
            return false;
        }

        var extended = await _coordinator.ExtendLockAsync(Key, Token, expiry);
        if (!extended)
        {
            IsAcquired = false;
        }

        return extended;
    }

    public async Task ReleaseAsync(CancellationToken cancellationToken = default)
    {
        if (_disposed || !IsAcquired)
        {
            return;
        }

        await _coordinator.ReleaseLockAsync(Key, Token);
        IsAcquired = false;
    }

    public async ValueTask DisposeAsync()
    {
        if (_disposed)
        {
            return;
        }

        if (IsAcquired)
        {
            await ReleaseAsync();
        }

        _disposed = true;
    }
}
